package com.katze.common.socket.client;

import com.katze.common.env.AppEnvironment;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.function.UnaryOperator;

public class SocketHeartBeatClient {
    private static final Logger log = LoggerFactory.getLogger(SocketHeartBeatClient.class);

    private final ScheduledExecutorService service = Executors.newScheduledThreadPool(0, new CustomizableThreadFactory("Socket-Scheduled-ThreadPool-"));

    private final InetSocketAddress address;

    private Supplier<String> active;

    public SocketHeartBeatClient(InetSocketAddress address) {
        this.address = address;
    }

    public void active(Supplier<String> active) {
        this.active = active;
    }

    public void initialization() {
        SocketHeartBeatClient.send(address, writer -> writer.append("online").append("@").println(active.get()), message -> {
            if ("online".equals(message)) {
                log.info("{}项目已上线", AppEnvironment.APP_NAME.getValue());
            }
            return message;
        });
    }

    public void heartbeat(Consumer<Boolean> call) {
        boolean isSuccess = SocketHeartBeatClient.send(address, writer -> writer.println("ping"), StringUtils::isNotBlank).apply((b) -> b);
        call.accept(isSuccess);
    }

    public void destroy() {
        SocketHeartBeatClient.send(address, writer -> writer.append("offline").append("@").println(active.get()), message -> {
            if ("offline".equals(message)) {
                log.info("{}项目已下线", AppEnvironment.APP_NAME.getValue());
            }
            return message;
        }).apply((s) -> {
            service.shutdown();
            return s;
        });
    }

    public static <T> Function<UnaryOperator<T>, T> send(InetSocketAddress address, Consumer<PrintWriter> writer, Function<String, T> message) {
        try (Socket socket = new Socket()) {
            socket.connect(address, 3000);
            try (BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
                 PrintWriter print = new PrintWriter(socket.getOutputStream(), true)) {
                writer.accept(print);
                T t = message.apply(reader.readLine());
                return (v) -> v.apply(t);
            }
        } catch (Exception e) {
            log.error("操作失败, 服务端：{}", address.toString(), e);
            return (v) -> v.apply(null);
        }
    }
}
